package functions;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Table;
import pojo.LogBean;
import pojo.ResultBean;
import pojo.RuleParam;
import ruleconstruct.Rulesimulator;
import service.*;

public class MyversionV2KeyedProcessFunc extends KeyedProcessFunction<String, LogBean, ResultBean> {
    transient ListState<LogBean> eventstate;
    UserProfileQueryService userProfileQueryService;
    UserActionSequenceQueryService userActionSequenceQueryService;
    UserActionCountQuerySerice userActionCountQuerySerice;
    RuleParam ruleParam;


    @Override
    public void open(Configuration parameters) throws Exception {
        eventstate = getRuntimeContext().getListState(new ListStateDescriptor<LogBean>("aa", LogBean.class));
        //接口引用指向实例对象
        userProfileQueryService = new UserProfileQueryServiceImpl();
        userActionCountQuerySerice = new UserActionCountQuerySericeImpl();
        userActionSequenceQueryService = new UserActionSequenceQueryServiceImpl();
        ruleParam = Rulesimulator.createrule();


    }

    @Override
    public void processElement(LogBean value, Context ctx, Collector<ResultBean> out) throws Exception {
        //更新状态
        eventstate.add(value);

        if (value.getEventId().equals(ruleParam.getTriggerparam().getEventId())) {

            boolean isuserprofilematch = userProfileQueryService.QueryUserProfileQueryFromHbase(value, ruleParam);

            if (!isuserprofilematch) return ;

            Boolean iscountMacth = userActionCountQuerySerice.UserActionCountQueryFromstate(eventstate, ruleParam);
            if(!iscountMacth)   return ;


            boolean isSequenceMatch = userActionSequenceQueryService.UserActionSequenceQueryFromState(eventstate, ruleParam);
            if (isSequenceMatch) {
                System.out.println("用户行为序列匹配成功");
            }

            ResultBean resultBean = new ResultBean();
            resultBean.setRuleid(ruleParam.getRuleId());
            resultBean.setDeviceid(value.getDeviceId());
            out.collect(resultBean);

        }
    }
}
